package RabbitMQ

import (
	"encoding/json"
	"fmt"
	"github.com/streadway/amqp"
	"hanmoon-product/datamodels"
	"hanmoon-product/services"
	"log"
	"sync"
)

const MQURL = "amqp://hanMoon:hanMoon@127.0.0.1:5672/hanMoon"

type RabbitMQ struct {
	conn    *amqp.Connection
	channel *amqp.Channel
	//队列名字
	QueueName string
	//交换机
	Exchange string
	//key
	Key string
	//连接信息
	Mqurl string
	sync.Mutex
}

//创建实例
func NewRabbitMQ(queueName string, exchange string, key string) *RabbitMQ {
	rabbitmq := &RabbitMQ{
		QueueName: queueName,
		Exchange:  exchange,
		Key:       key,
		Mqurl:     MQURL}
	var err error
	//创建rabbitmq连接
	rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
	rabbitmq.failOnErr(err, "创建连接错误")
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	rabbitmq.failOnErr(err, "创建Channel失败")
	return rabbitmq
}

//断开连接
func (r *RabbitMQ) Destory() {
	r.channel.Close()
	r.conn.Close()
}

//错误处理函数
func (r *RabbitMQ) failOnErr(err error, message string) {
	if err != nil {
		log.Fatalf("%S:%S", message, err)
		panic(fmt.Sprintf("%S:%S", message, err))
	}
}

//Step 1:Simple 模式 创建实例
func NewRabbitSimple(queueName string) *RabbitMQ {
	rabbitmq := NewRabbitMQ(queueName, "", "")
	return rabbitmq
}

//Step 2:Simple下生产 代码
func (r *RabbitMQ) PulishSimple(message string) error {
	r.Lock()
	defer r.Unlock()
	//1.申请队列
	_, err := r.channel.QueueDeclare(
		r.QueueName,
		false,
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		return err
	}
	//2.发送信息
	r.channel.Publish(
		r.Exchange,
		r.QueueName,
		false,
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
	return nil
}

//Step 3:Simple下消费 代码
func (r *RabbitMQ) ConsumeSimple(orderService services.IOrderService,
	productService services.IProductService) {
	//1.申请队列
	_, err := r.channel.QueueDeclare(
		r.QueueName,
		false,
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		fmt.Println(err)
	}

	//防止爆库
	r.channel.Qos(
		1, //当前消费者一笔接受最大消费者数量
		0,
		false, //为true表示channel全局可用
	)

	//2.消费消息
	msgs, err := r.channel.Consume(
		r.QueueName,
		"",
		false,
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		fmt.Println(err)
	}
	forever := make(chan bool)

	//3.使用协程消费
	go func() {
		for d := range msgs {
			log.Printf("Received a message:%s", d.Body)
			message := &datamodels.Message{}
			err := json.Unmarshal([]byte(d.Body), message)
			if err != nil {
				fmt.Println(err)
			}
			_, err = orderService.InsertOrderBymessage(message)
			if err != nil {
				fmt.Println(err)
			}
			err = productService.SubNumberOne(message.ProductID)
			if err != nil {
				fmt.Println(err)
			}
			//手动应答，如果为true表示确认所有被确认的消息，
			//为false确认当前消息
			d.Ack(false)
		}
	}()

	log.Printf("[ * ] Waiting for messages,to exit press Ctrl+C")

	<-forever
}

//Step 1:PubSub 模式 创建实例
func NewRabbitPubSub(exchangeName string) *RabbitMQ {
	rabbitmq := NewRabbitMQ("", exchangeName, "")
	var err error
	//创建rabbitmq连接
	rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
	rabbitmq.failOnErr(err, "创建连接错误")
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	rabbitmq.failOnErr(err, "创建Channel失败")
	return rabbitmq
}

//Step 2:PubSub下生产 代码
func (r *RabbitMQ) PulishPubSub(message string) {
	//1.尝试创建交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		"fanout",
		true,
		false,
		false,
		false,
		nil)
	r.failOnErr(err, "Failed to declare an exchange")
	//2.发送信息
	r.channel.Publish(
		r.Exchange,
		"",
		false,
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
}

//Step 3:PubSub下消费 代码
func (r *RabbitMQ) RecieveSub() {
	//1.尝试创建交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		"fanout",
		true,
		false,
		false,
		false,
		nil)
	r.failOnErr(err, "Failed to declare an exchange")
	//2.试探性创建队列
	q, err := r.channel.QueueDeclare(
		"",
		false,
		false,
		true,
		false,
		nil)
	r.failOnErr(err, "Failed to declare a queue")
	//3.绑定队列到交换机
	r.channel.QueueBind(
		q.Name,
		"",
		r.Exchange,
		false,
		nil)

	//4.消费信息
	msgs, err := r.channel.Consume(
		q.Name,
		"",
		true,
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		fmt.Println(err)
	}
	forever := make(chan bool)

	//3.使用协程消费
	go func() {
		for d := range msgs {
			log.Printf("Received a message:%s", d.Body)
		}
	}()

	log.Printf("[ * ] Waiting for messages,to exit press Ctrl+C")

	<-forever
}

//Step 1: 路由模式 创建实例
func NewRabbitRoutind(exchangeName string, routingKey string) *RabbitMQ {
	rabbitmq := NewRabbitMQ("", exchangeName, routingKey)
	var err error
	//创建rabbitmq连接
	rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
	rabbitmq.failOnErr(err, "创建连接错误")
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	rabbitmq.failOnErr(err, "创建Channel失败")
	return rabbitmq
}

//Step 2:路由模式下生产 代码
func (r *RabbitMQ) PulishRouting(message string) {
	//1.尝试创建交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		"direct",
		true,
		false,
		false,
		false,
		nil)
	r.failOnErr(err, "Failed to declare an exchange")
	//2.发送信息
	r.channel.Publish(
		r.Exchange,
		r.Key,
		false,
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
}

//Step 3:路由模式下消费 代码
func (r *RabbitMQ) RecieveRouting() {
	//1.尝试创建交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		"direct",
		true,
		false,
		false,
		false,
		nil)
	r.failOnErr(err, "Failed to declare an exchange")
	//2.试探性创建队列
	q, err := r.channel.QueueDeclare(
		"",
		false,
		false,
		true,
		false,
		nil)
	r.failOnErr(err, "Failed to declare a queue")
	//3.绑定队列到交换机
	r.channel.QueueBind(
		q.Name,
		r.Key,
		r.Exchange,
		false,
		nil)

	//4.消费信息
	msgs, err := r.channel.Consume(
		q.Name,
		"",
		true,
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		fmt.Println(err)
	}
	forever := make(chan bool)

	//3.使用协程消费
	go func() {
		for d := range msgs {
			log.Printf("Received a message:%s", d.Body)
		}
	}()

	log.Printf("[ * ] Waiting for messages,to exit press Ctrl+C")

	<-forever
}
